package com.yunban.project.api.text_check.utils;

import cn.hutool.crypto.digest.MD5;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.yunban.project.api.text_check.domain.WordParseRecord;
import com.yunban.project.api.text_check.service.impl.DeepSeekServiceImpl;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import net.lab1024.sa.base.common.exception.BusinessException;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author: lijiajun
 * @date: 2025-03-03
 * @version: 1.0
 */
@Slf4j
@Data
public class SSEHandler {

//    public static void main(String[] args) {
//        System.out.println(UUID.randomUUID().toString());
//        // 通过 userId 生成一个 24 位字符串
//        // 1L  c4ca4238a0b923820dcc509a6f75849b
//        // c81e728d9d4c2f636f067f89cc14862c
//        long userId = 2000L;
//        System.out.println();
//    }

    /**
     * 异步处理SSE流并返回Mono<String>表示最终HTML结果
     */
    public static Mono<String> tencentSSEChat(String url, String prompt, String botAppKey, String taskId) {
        DeepSeekServiceImpl.TencentSSEChatReq req = new DeepSeekServiceImpl.TencentSSEChatReq(prompt, botAppKey);
        // 请求 ID，用于标识一个请求（作消息串联，建议每个请求使用不同的 request_id）
        req.setRequest_id(UUID.randomUUID().toString());
        // 会话 ID，用于标识一个会话（外部系统提供，建议不同的用户端会话传入不同的 session_id，否则同一个应用下的不同用户的消息记录会串掉）
        req.setSession_id(UUID.randomUUID().toString());
//        req.setSession_id(MD5.create().digestHex(String.valueOf(taskId), "UTF-8"));
        // 访客 ID（外部输入，建议唯一，标识当前接入会话的用户）
        req.setVisitor_biz_id(UUID.randomUUID().toString());
//        req.setVisitor_biz_id(String.valueOf(taskId));
        req.setStreaming_throttle(5);

        WordParseRecord parseRecord = new WordParseRecord();
        WebClient webClient = WebClient.create();
        return webClient.post()
                .uri(url)
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(req)
                .accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve()
                .bodyToFlux(String.class)
                .doOnNext(data -> processSSEData(data, parseRecord))
                .then(Mono.defer(() -> {
                    String htmlResult = parseRecord.getHtml();
                    if (htmlResult == null || htmlResult.isEmpty()) {
                        return Mono.error(new IllegalStateException("No result received"));
                    }
                    return Mono.just(htmlResult);
                }));
    }

    private static void processSSEData(String data, WordParseRecord parseRecord) {
        try {
            log.info("Received SSE data:{}", data);
            JSONObject sseJson = JSON.parseObject(data);
            JSONObject payloadJson = JSON.parseObject(JSON.toJSONString(sseJson.get("payload")));

            Object isFinalObj = payloadJson.get("is_final");
            if (Objects.isNull(isFinalObj)) {
                return;
            }
            boolean isFinal = Boolean.parseBoolean(isFinalObj.toString());
            if (isFinal && "reply".equals(sseJson.getString("type"))) {
                String content = payloadJson.getString("content");
                parseRecord.setHtml(content);
                log.info("Final result: " + content);
            }
        } catch (Exception ex) {
            System.err.println("Error processing SSE data: " + ex.getMessage());
            log.error("Error processing SSE data: " + ex.getMessage());
            throw new BusinessException("Error processing SSE data: " + ex.getMessage());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        String text = "我<error-text :金天>今天</error-text>来到了浦东市团委";
        CountDownLatch latch = new CountDownLatch(1);
        AtomicReference<String> resultHolder = new AtomicReference<>();

        tencentSSEChat("https://wss.lke.cloud.tencent.com/v1/qbot/chat/sse", text, "lsWjhbZT", "23")
                .subscribe(result -> {
                    resultHolder.set(result);
                    latch.countDown(); // 计数器减1，表示处理完成
                }, error -> {
                    log.error("Error occurred: " + error.getMessage());
                    latch.countDown(); // 计数器减1，即使发生错误也要退出
                });
        // 主线程等待异步处理完成
        latch.await();
        // 返回结果
        String finalResult = resultHolder.get();
        if (finalResult != null) {
            System.out.println("Final result in main thread: " + finalResult);
        } else {
            System.err.println("No result received.");
        }
    }
}
